package com.telra.app.server.cassandra;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import com.telra.app.server.protobuff.QueryRequestProto.QueryRequest;
import com.telra.app.server.protobuff.OrderProtos.OrderTicket;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;


import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SliceQuery;

public class CassandraQuery {

    private Cluster cluster;
    private Keyspace keyspace;
    private static StringSerializer stringSerializer = StringSerializer.get();
    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(CassandraQuery.class);

    private CassandraQuery() {
        //get the cluster
        cluster = HFactory.getOrCreateCluster("TestCluster", "localhost:9160");
        //get the keyspace
        keyspace = HFactory.createKeyspace("Keyspace1", cluster);

        // Create a ConnectionFactory
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        try {
            // Create a Connection
            Connection connection = connectionFactory.createConnection();
            connection.start();

            // Create a Session
            final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            final MessageProducer replyProducer = session.createProducer(null);
            replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            MessageConsumer consumer = session.createConsumer(session.createQueue("QUERY.OUT"));
            consumer.setMessageListener(new MessageListener() {

                public void onMessage(final Message m) {

                    log.info("query request received: " + m);

                    try {

                        ObjectMessage bMessage = (ObjectMessage) m;
                        QueryRequest o = (QueryRequest)bMessage.getObject();

                        queryOrderByOrderId(o.getOrderidKey(),
                                new OrderQueryResult() {
                                    public void result(Map<?, ?> map) {
                                        try {
                                            Destination dest = m.getJMSReplyTo();
                                            ObjectMessage bMessage = session.createObjectMessage();
                                            //TODO Convert Map result to protobuf
                                            OrderTicket o = OrderTicket.newBuilder().
                                                    setItemCategory(OrderTicket.Category.ACCESSORIES).
                                                    setItemName(map.get("itemName").toString()).
                                                    setPrice(Double.valueOf(map.get("price").toString())).build();
                                            bMessage.setObject(o);
                                            replyProducer.send(dest, bMessage);
                                            log.info("sent query result" + o);
                                        } catch (JMSException e) {
                                            e.printStackTrace();
                                        }
                                    }
                                }
                        );

                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

        }
        catch (JMSException e) {
            e.printStackTrace();
        }

    }

    public void queryOrderByOrderId(String orderID, final OrderQueryResult result) {

        final SliceQuery<String, String, String> q =
                HFactory.createSliceQuery(keyspace, stringSerializer, stringSerializer, stringSerializer);

        q.setColumnFamily("Standard1").setKey(orderID).
                setColumnNames("category","price","itemName");

        Runnable r = new Runnable() {
            public void run() {
                QueryResult<ColumnSlice<String,String>> r = q.execute();

                List<HColumn<String,String>> colList = r.get().getColumns();
                Map<String, String> map = new HashMap<String, String>();

                for (HColumn<String,String> col : colList) {
                    map.put(col.getName() , col.getValue());
                }

                result.result(map);
            }
        };

        new Thread(r).start();

    }

    public static CassandraQuery getInstance() {
        return new CassandraQuery();
    }

    public static void main(String args[]) {

        CassandraQuery.getInstance();

        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        connectionFactory.setUseAsyncSend(false);	//blocking call
        Connection connection;
        try {
            connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination tempDest = session.createTemporaryQueue();

            QueryRequest req = QueryRequest.newBuilder().setOrderidKey("1304126165018").build();
            MessageProducer producer = session.createProducer(session.createQueue("QUERY.OUT"));
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            ObjectMessage bMessage = session.createObjectMessage();
            bMessage.setObject(req);
            bMessage.setJMSReplyTo(tempDest);

            producer.send(bMessage);



        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        log.info("send query request");


    }

}
